package com.navi.loaddata.loadService.impl.shtWipLoad;


import com.navi.loaddata.core.etlException.ETLException;
import com.navi.loaddata.core.gpEntity.WppShtOpeConf;
import com.navi.loaddata.core.mysqlEntity.EtlConfD;
import com.navi.loaddata.core.repository.impl.GreenPlumRepositoryImpl;
import com.navi.loaddata.core.util.LoggerUtil;
import org.hibernate.SQLQuery;
import org.hibernate.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

@Service
@Transactional(transactionManager = "secondaryPlatformTransactionManager",
        rollbackFor = {RuntimeException.class, Exception.class})
public class LoadShtWipOpeService {

    private static final Logger logger = LoggerFactory.getLogger(LoadShtWipTmpService.class);
    private LoggerUtil loggerUtil = new LoggerUtil(this.getClass());

    @Autowired
    private GreenPlumRepositoryImpl gpDao;

    public void loadMain1(EtlConfD etlConfD) throws ETLException {
        Session session = gpDao.getEntityManagerSession();
        try {
            logger.info("Load data  To Tmp start");
            logger.info("etlConfD message : etlconfd startTime: "+etlConfD.getRunStartTimestamp()+
                    "---etlconfd endTime:  "+etlConfD.getRunEndTimestamp() +"---etlconfd ext1"+etlConfD.getExt1());
        //    loadDataToInfoTmp(session, etlConfD);   // hisAndExtToTmp
            loadDataToOpeInfoTmp(session, etlConfD);
            logger.info("Load data  To Tmp finish");
        } catch (ETLException e) {
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException(e.getMessage());
        }
    }

    public void loadMain2(EtlConfD etlConfD) throws ETLException {
        Session session = gpDao.getEntityManagerSession();
        try {
            logger.info("Load data  To TmpSum start");
    //        loadDataTmpToInfo(session, etlConfD);
            loadDatOpeTmpToOpeInfo(session, etlConfD);
            logger.info("Load data  To TmpSum finish");
            logger.info("etlConfD message : etlconfd startTime: "+etlConfD.getRunStartTimestamp()+
                    "---etlconfd endTime:  "+etlConfD.getRunEndTimestamp() +"---etlconfd ext1"+etlConfD.getExt1());
        } catch (ETLException e) {
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException(e.getMessage());
        }
    }

    private void tmpToDistinct(Session session, List<String> tableColumns,String tmpTableName,String distinctTableName) throws ETLException {
        //tmpTableName ="wpp_sht_info_tmp";
        //distinctTableName="wpp_sht_info_distinct";
        StringBuffer sqlbuffDistinct = new StringBuffer();
        try {
            //存储中间结果
            sqlbuffDistinct.append(" INSERT INTO STAGE."+distinctTableName+" (panel_id  ");
            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffDistinct.append("," + column);
            });
            sqlbuffDistinct.append(" ) ");

            sqlbuffDistinct.append(" SELECT DISTINCT ON (PANEL_ID) ");
            sqlbuffDistinct.append("  PANEL_ID");
            //对每一字段做更新处理，若为空，保留原值
            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffDistinct.append(", first_value(" + column + " ) over (partition by panel_id order by ");
//            sqlbuffDistinct.append( " (CASE WHEN "+column+" IS NULL THEN NULL ELSE db_timestamp END) DESC NULLS LAST) AS "+column  );
//                sqlbuffDistinct.append("db_timestamp  DESC ) AS " + column);
                sqlbuffDistinct.append( " (CASE WHEN "+column+" IS NULL THEN NULL ELSE db_timestamp END)," );
                sqlbuffDistinct.append( column+" is not null DESC ) AS "+column  );

            });
            sqlbuffDistinct.append("  FROM stage."+tmpTableName+"  where 1=1");

            SQLQuery queryLoadStep = session.createSQLQuery(sqlbuffDistinct.toString());
            int count =   queryLoadStep.executeUpdate();
            logger.info(tmpTableName+"  To  "+distinctTableName+"  count :" +count);
        } catch (Exception e) {
            logger.error("************   sqlbuffDistinct.toString():  **********    "+sqlbuffDistinct.toString());
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException("ERROR tmpToDistinct INSERT:" + e.getMessage());
        }

    }


    private void distinctToMerge(Session session, List<String> tableColumns,String distinctTableName,String mergeTableNmae,String finalTableName) throws ETLException {
        //distinctTableName="wpp_sht_info_distinct";
        //mergeTableNmae="wpp_sht_info_merge";
        //finalTableName="wpp_sht_info";
        //合并写入
        StringBuffer sqlbuffMerge = new StringBuffer();
        try {

            sqlbuffMerge.append(" INSERT INTO STAGE."+mergeTableNmae+" (panel_id  ");

            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffMerge.append("," + column);
            });
            sqlbuffMerge.append(" ) ");
            sqlbuffMerge.append(" SELECT D.PANEL_ID ");
            //对每一字段做更新处理
            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffMerge.append(", coalesce(D." + column + ", I." + column + ") " + column);
            });
            sqlbuffMerge.append("  FROM STAGE."+distinctTableName+" AS  D  LEFT JOIN  "+finalTableName+"  I  USING (PANEL_ID)");  //left join ? inner join

            SQLQuery queryLoadStep = session.createSQLQuery(sqlbuffMerge.toString());
            int count =   queryLoadStep.executeUpdate();
            logger.info(distinctTableName+"  To  "+mergeTableNmae+"  count :" +count);
        } catch (Exception e) {
            logger.error("************   sqlbuffMerge.toString():  **********    "+sqlbuffMerge.toString());
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException("ERROR tmpToDistinct INSERT:" + e.getMessage());
        }
    }

    private void mergeToInfo(Session session, List<String> tableColumns,String mergeTableName,String finalTableName) throws ETLException {
        // mergeTableName ="WPP_SHT_INFO";
        //finalTableName ="WPP_SHT_INFO_MERGE";
        String sqlDelete = " DELETE FROM "+finalTableName+"  I USING STAGE."+mergeTableName+" AS M  WHERE I.PANEL_ID=M.PANEL_ID ";
        StringBuffer sqlbuffInfo = new StringBuffer();
        try {
            // DELETE USING，删除全量表的符合条件的记录。
            SQLQuery queryLoadStepD = session.createSQLQuery(sqlDelete);
            queryLoadStepD.executeUpdate();
            //  INSERT wpp_sht_info

            sqlbuffInfo.append(" INSERT INTO "+finalTableName+" (panel_id");
            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffInfo.append("," + column);
            });
            sqlbuffInfo.append(" ) ");
            //  sqlbuffInfo.append(" SELECT D.* FROM WPP_SHT_INFO_DISTINCT D LEFT JOIN STAGE.WPP_SHT_INFO_MERGE M USING (PANEL_ID) WHERE M.* IS NULL   " );
            //  sqlbuffInfo.append(" UNION ALL   "   );
            sqlbuffInfo.append("  SELECT panel_id");
            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffInfo.append("," + column);
            });
            sqlbuffInfo.append("  FROM STAGE."+mergeTableName);
            SQLQuery queryLoadStep = session.createSQLQuery(sqlbuffInfo.toString());
            int count =   queryLoadStep.executeUpdate();
            logger.info(mergeTableName+"  To  "+finalTableName+" count :" +count);
        } catch (Exception e) {
            logger.error("************   sqlDelete.toString():  **********    "+sqlDelete.toString());
            logger.error("************   sqlbuffInfo.toString():  **********    "+sqlbuffInfo.toString());
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException("ERROR mergeToInfo :" + e.getMessage());
        }


    }

    //清空临时表
    private void empotyTmp(Session session, String tableName) throws ETLException {
        StringBuffer sqlbuff = new StringBuffer();
        try {
            sqlbuff.append(" TRUNCATE TABLE " + tableName);
            SQLQuery queryLoadStep = session.createSQLQuery(sqlbuff.toString());
            int count =   queryLoadStep.executeUpdate();
            logger.info("TRUNCATE table :"+tableName+"   count :" +count);
        } catch (Exception e) {
            logger.error("************   empotyTmp.toString():  **********    "+sqlbuff.toString());
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException("ERROR TRUNCATE TABLE " + tableName+" :" + e.getMessage());
        }
    }


    /**
     *Load OpeInfoTmp
     */


    private void loadDataToOpeInfoTmp(Session session, EtlConfD etlConfD) throws ETLException {
        String tableA = "wpp_asht_ope_his";
        String tableC = "wpp_csht_ope_his";
        String tableF = "wpp_fsht_ope_his";
        loadDataToOpeInfoTmpX(session, etlConfD,"ARRAY", tableA);
        loadDataToOpeInfoTmpX(session, etlConfD,"OC", tableC);
        loadDataToOpeInfoTmpX(session, etlConfD,"CF", tableF);

    }

    private  void loadDataToOpeInfoTmpX(Session session, EtlConfD etlConfD,String shopId,String tableName) throws ETLException {
        try {
            //启动加载ope_conf
            // shopId = "ARRAY";
            //  tableName = "wpp_asht_ope_his";
            String tmpTableName="wpp_sht_ope_info_tmp";
            List<WppShtOpeConf> opeConfList = gpDao.list(WppShtOpeConf.class, "from WppShtOpeConf where valigFlg = 'Y' and shopId='" + shopId + "'");
            List<String> tableColumns = new ArrayList<>();
            /*try {
                tableColumns = gpDao.findBySQL(" SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = '" + tmpTableName + "' ");  //获取表中所有列名
            } catch (Exception e) {
                logger.error(LoggerUtil.stackTraceToString(e));
                throw new ETLException("ERROR loadDataToOpeInfoTmp GET COLUMN_NAME:" + e.getMessage());
            }*/
            HashMap<String, String> opeConfMap = new HashMap<>();
            StringBuffer sqlbuff = new StringBuffer();
            sqlbuff.append(" select H.glass_id as panel_id ");
            //行转列
            opeConfList.forEach(ope -> {
                String opeId = ope.getOpeId();
                String targetSeq = ope.getTargetSeq();
                sqlbuff.append(" ,MAX(case when ope_id = '" + opeId + "' then logon_timestamp else null end) as move_in_time_" + targetSeq);
                sqlbuff.append(" ,MAX(case when ope_id = '" + opeId + "' then logof_timestamp else null end) as move_out_time_" + targetSeq);
                sqlbuff.append(" ,MAX(case when ope_id = '" + opeId + "' then recipe_id else null end) as recipe_id_" + targetSeq);
                sqlbuff.append(" ,MAX(case when ope_id = '" + opeId + "' then eqpt_id else null end) as eqpt_id_" + targetSeq);
                tableColumns.add("move_in_time_" + targetSeq);
                tableColumns.add("move_out_time_" + targetSeq);
                tableColumns.add("recipe_id_" + targetSeq);
                tableColumns.add("eqpt_id_" + targetSeq);
            });
            sqlbuff.append("  FROM " + tableName + " H  where ");
            // sqlbuff.append("  H.db_timestamp between '2017-11-02 15:14:37' and '2018-04-02 20:14:37'");
            sqlbuff.append("   H.DB_TIMESTAMP BETWEEN '" + etlConfD.getRunStartTimestamp() + "' AND '" + etlConfD.getRunEndTimestamp() + "'");

            // sqlbuff.append(" and H.glass_id ='A17CL01EBC' ");
            sqlbuff.append(" group by H.glass_id ");
            StringBuffer sqlbuffSum = new StringBuffer();
            sqlbuffSum.append(" INSERT INTO STAGE."+tmpTableName+" (panel_id  ");
            //获取所有列
            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffSum.append("," + column);
            });

            sqlbuffSum.append(")").append(sqlbuff);
            SQLQuery queryLoadStep = session.createSQLQuery(sqlbuffSum.toString());
            int count =   queryLoadStep.executeUpdate();
            logger.info(tableName +"----loadDataToOpeInfoTmp count :" +count);
        }catch (Exception e) {
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException("ERROR loadDataToOpeInfoTmp :" + e.getMessage());
        }
    }

    public int loadDatOpeTmpToOpeInfo(Session session, EtlConfD etlConfD) throws ETLException {
        int countSize = 0;
        try {
            empotyTmp(session, "stage.wpp_sht_ope_info_merge");
            empotyTmp(session, "stage.wpp_sht_ope_info_distinct");
            String tableName = "wpp_sht_ope_info";
            StringBuffer sqlbuffColumn = new StringBuffer();
            sqlbuffColumn.append(" SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = '" + tableName + "' " +
                    " ORDER BY ORDINAL_POSITION;");
            List<String> tableColumns = gpDao.findBySQL(sqlbuffColumn.toString());  //获取表中所有列名
            tmpToDistinct(session, tableColumns,"wpp_sht_ope_info_tmp","wpp_sht_ope_info_distinct");
            distinctToMerge(session, tableColumns,"wpp_sht_ope_info_distinct","wpp_sht_ope_info_merge","wpp_sht_ope_info");
            mergeToInfo(session, tableColumns,"wpp_sht_ope_info_merge","wpp_sht_ope_info");

            List<String> distinct = gpDao.findBySQL(" SELECT d.panel_id FROM stage.wpp_sht_ope_info_distinct d where 1=1");
            List<String> merges = gpDao.findBySQL(" SELECT m.panel_id FROM stage.wpp_sht_ope_info_merge m where 1=1");
            empotyTmp(session, "stage.wpp_sht_ope_info_merge");
            empotyTmp(session, "stage.wpp_sht_ope_info_distinct");
//            logger.info("delete over :merges.size()=" + merges.size() + "---distinct.size()=" + distinct.size());
            if (merges.size() > 0 && distinct.size() > 0) {
                countSize = distinct.size();
                empotyTmp(session, "stage.wpp_sht_ope_info_tmp");
                logger.info(" 清空 distinct，merge，tmp");
            } else {
                logger.info(" 只清空 distinct，merge，不清空OpeInfotmp");
            }
            updateWppShtInfoData(session);
        } catch (Exception e) {
            logger.error(LoggerUtil.stackTraceToString(e));
            if ("ERROR".equals(e.getMessage().substring(0, 5))) {
                throw e;
            }
            throw new ETLException("ERROR loadDatOpeTmpToOpeInfo INSERT:" + e.getMessage());
        }
        return countSize;


    }

    /**
     *    对 wpp_sht_ope_info  表进行大板，小板 数据互补
     */
    private void updateWppShtInfoData(Session session)throws ETLException{
        StringBuffer sqlbuff = new StringBuffer();
        try {
//用ARRAY补全wpp_sht_ope_info  OC数据
            String sqlAdd="";
            sqlbuff.append(" update wpp_sht_ope_info o set    ");
            for(int i=1;i<=100;i++){
                String tag;
                if(i<10){
                    tag= ""+"00"+i;
                }else if(i>=10 && i<100){
                    tag= ""+"0"+i;
                }else{
                    tag= ""+i;
                }
                sqlbuff.append(" eqpt_id_"+tag+"           =case when a.eqpt_id_"+tag+"$           is not null then a.eqpt_id_"+tag+"$            else eqpt_id_"+tag+" end, ");
                sqlbuff.append(" move_in_time_"+tag+"      =case when a.move_in_time_"+tag+"$      is not null then a.move_in_time_"+tag+"$       else move_in_time_"+tag+" end, ");
                sqlbuff.append(" recipe_id_"+tag+"         =case when a.recipe_id_"+tag+"$         is not null then a.recipe_id_"+tag+"$          else recipe_id_"+tag+" end, ");
                sqlbuff.append(" move_out_time_"+tag+"     =case when a.move_out_time_"+tag+"$     is not null then a.move_out_time_"+tag+"$      else move_out_time_"+tag+" end,");
                sqlAdd+=" ,h.eqpt_id_"+tag+" as eqpt_id_"+tag+"$,h.move_in_time_"+tag+" as move_in_time_"+tag+"$," +
                        " h.recipe_id_"+tag+" as recipe_id_"+tag+"$,h.move_out_time_"+tag+" as move_out_time_"+tag+"$  ";
            }
            sqlbuff.deleteCharAt(sqlbuff.length()-1);
            sqlbuff.append(" from( select i.*,h.panel_id ");
            //sqlbuff.append(" ,h.eqpt_id_007 as eqpt_id_007$,h.move_in_time_007 as move_in_time_007$ ");
            sqlbuff.append(sqlAdd);
            sqlbuff.append(" from (select panel_id as panel,tft_glass_id,cf_glass_id from wpp_sht_info )i ");
            sqlbuff.append(" left join (select * from wpp_sht_ope_info where panel_id in ");
            sqlbuff.append("  (select glass_id from  wpp_asht_ope_his )  )h  on ");
            sqlbuff.append(" i.tft_glass_id =h.panel_id where panel_id is not null and panel_id<>panel  ");
            sqlbuff.append(" and length(panel)>10 )a where  o.panel_id =a.panel; ");
//用CF补全wpp_sht_ope_info  OC数据
            String sqlFdd="";
            sqlbuff.append(" update wpp_sht_ope_info o set    ");
            for(int i=1;i<=100;i++){
                String tag;
                if(i<10){
                    tag= ""+"00"+i;
                }else if(i>=10 && i<100){
                    tag= ""+"0"+i;
                }else{
                    tag= ""+i;
                }
                sqlbuff.append(" eqpt_id_"+tag+"      =case when eqpt_id_"+tag+"      is null then a.eqpt_id_"+tag+"$       else eqpt_id_"+tag+" end, ");
                sqlbuff.append(" move_in_time_"+tag+"      =case when move_in_time_"+tag+"      is null then a.move_in_time_"+tag+"$       else move_in_time_"+tag+" end, ");
                sqlbuff.append(" recipe_id_"+tag+"     =case when recipe_id_"+tag+"      is null then a.recipe_id_"+tag+"$       else recipe_id_"+tag+" end, ");
                sqlbuff.append(" move_out_time_"+tag+"     =case when move_out_time_"+tag+"      is null then a.move_out_time_"+tag+"$       else move_out_time_"+tag+" end,");
                sqlFdd+=" ,h.eqpt_id_"+tag+" as eqpt_id_"+tag+"$,h.move_in_time_"+tag+" as move_in_time_"+tag+"$," +
                        " h.recipe_id_"+tag+" as recipe_id_"+tag+"$,h.move_out_time_"+tag+" as move_out_time_"+tag+"$  ";
            }
            sqlbuff.deleteCharAt(sqlbuff.length()-1);
            sqlbuff.append(" from( select i.*,h.panel_id ");
            //sqlbuff.append(" ,h.eqpt_id_007 as eqpt_id_007$,h.move_in_time_007 as move_in_time_007$ ");
            sqlbuff.append(sqlFdd);
            sqlbuff.append(" from (select panel_id as panel,tft_glass_id,cf_glass_id from wpp_sht_info )i ");
            sqlbuff.append(" left join (select * from wpp_sht_ope_info where panel_id in ");
            sqlbuff.append("  (select glass_id from  wpp_fsht_ope_his )  )h  on ");
            sqlbuff.append("i.cf_glass_id =h.panel_id where panel_id is not null and panel_id<>panel  ");
            sqlbuff.append(" and length(panel)>10 )a where  o.panel_id =a.panel; ");

            long startTimes = System.currentTimeMillis();
            SQLQuery queryLoadStep = session.createSQLQuery(sqlbuff.toString());
            int count =   queryLoadStep.executeUpdate();
            long endTimes = System.currentTimeMillis();
            long minutes = (endTimes - startTimes) / 60;
            logger.info("用CF大板补全ARRAY,OC板中CF数据;用 ARRAY大板数据  补全OC，CF板的ARRAY数据;  minutes:{}s---count{}:", minutes,count);
        } catch (Exception e) {
            logger.error("************  updateWppShtOpeData SQL:  **********    "+sqlbuff.toString());
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException("ERROR updateWppShtOpeData :" + e.getMessage());
        }
    }

}
